/**
 * ======================================================================
 * 【基本说明】：
 *  MobileIMSDK-Web版服务端的MQ消息中间件服务提供者实现类。
 *
 *  【本文件中实现的功能有】：
 *  1）一个消息生产者；
 *  2）一个消息消费者；
 *  3）（与MQ服务器）断线重连和恢复能力；
 *  [4）发送出错暂存到缓存数组（内存中），并在下次重连正常时自动重发；]
 *  5）遵照官方的最佳实践：一个连接（connection）、两个channel（一个用于生产者、一个用于消费者）；
 *  6）消费者手动ACK能力：业务层处理不成功可重新放回队列。
 *
 * 【基本用法】：
 * var imMQProvider = require('./im-mq-provider');
 *
 * Created by Jack Jiang on 16-9-10.
 * ======================================================================
 */


var amqp = require('amqplib/callback_api');
var config = require('../j_conf/config');
var logger = require('../j_conf/log').logger;
// MQ连接是否“正在”或“已经”start()，本标识用于防止在断线重连时出现两个循环调用start的线程
var running = false;
// if the connection is closed or fails to be established at all, we will reconnect
var amqpConn = null;
var pubChannel = null;

/* ##### 20160919 by js：为了让publish调用时能准确地反应出是否发送成功（而不需要在没发送成功时自动再次发送）
   #####                 ，暂时关闭了此离线再发功能，也许以后的其它逻辑中用的上，但还是暂时取消了相关代码！！
   #####                 【JS认为】桥接转发如不成功就应该立即由调用者知道，不然有此离线功能的话就会导致为何消息
   #####                 报已发出，但就是没有收到的问题，这对调试和用户体验都不利！*/
//var offlinePubQueue = [];

// MQ中间件收到桥接消息时的回调函数
var callBack_forWork = null;


/**
 * MQ中间件服务提供者的启动方法。
 */
function start() {
    // 防止被重复运行（根据下面的产生err即重新start的代码逻辑，如无本判断则会发生重启重启哦，不信仔细看代码！）
    if(running) {
        logger.warn('[3-IMMQ] - 【注意】MQ服务正在或已经start()，本次start()将被忽略！running=%s', running);
        return;
    }

    // 设置“正在”或“已经”start()标识
    running = true;

    logger.debug('[3-IMMQ] - [start()中] 要连接的MQ服务器地址：%s', config.im_mq_server_url + "?heartbeat=30");

    amqp.connect(config.im_mq_server_url + "?heartbeat=30", function(err, conn) {
        if (err) {
            running = false;
            logger.error('[3-IMMQ] - [start()中] amqp.connect出错了，回调中err.message：%s 【3秒后重新start重试】', err.message);
            return setTimeout(start, 3000);
        }

        // 连接成功后注册error事件监听
        conn.on("error", function(err) {
            logger.error('[3-IMMQ] - [start()中] conn.on("error")事件触发了，err：%s ', JSON.stringify(err));

            if (err.message !== "Connection closing") {
                console.error("[3-AMQP] - [start()中] conn error", err.message);
            }
        });

        // 连接成功后注册close事件监听
        conn.on("close", function() {
            running = false;
            logger.error('[3-IMMQ] - [start()中] conn.on("close")事件触发了，表示连接已关闭！【3秒后重新start重试】');
            return setTimeout(start, 3000);
        });

        logger.info("[3-IMMQ] - [start()中] 与RabittMQ服务器的连接成功完成！【OK】");
        amqpConn = conn;

        // 连接成功后要做的事
        whenConnected();
    });
}

function whenConnected() {
    startPublisher();
    startWorker();
}

function startPublisher() {
    if (amqpConn) {
        amqpConn.createConfirmChannel(function (err, ch) {
            if (closeOnErr(err, ' ↑[startPublisher()中的amqpConn.createConfirmChannel]')) return;

            ch.on("error", function (err) {
                logger.error('[3-IMMQ-↑] - [startPublisher()中] ch.on("error")事件触发了，err：%s ', JSON.stringify(err));
            });
            ch.on("close", function () {
                logger.error('[3-IMMQ-↑] - [startPublisher()中] ch.on("close")事件触发了，channel 关闭了！');
                //console.log("[AMQP] channel closed");
            });

            pubChannel = ch;

            //logger.info('[3-IMMQ-↑] - [startPublisher()中] 的channel成功创建了，马上开始循环publish消息，当前数组队列长度：%d！【OK】'
            //    , offlinePubQueue.length);
            logger.info('[3-IMMQ-↑] - [startPublisher()中] 的channel成功创建了，接下来的publish将能正常使用！【OK】');

            // 确保在publish之前，队列已经被建立，否则publish的消息都将无法被消费者收到（也不会被mq持久化，直接没了！）
            ch.assertQueue(config.im_mq_queue_web2app, { durable: true }, function(err, _ok) {
                if (closeOnErr(err, ' - ↑[[startPublisher(config.im_mq_queue_web2app)中] 的ch.assertQueue]')) return;

                logger.info('[3-IMMQ-↑] - [startPublisher(config.im_mq_queue_web2app)中] Queue已成功建立，Publisher初始化成功，消息将可publish过去且不怕丢失了。【OK】');

                /* ##### 20160919 by js：为了让publish调用时能准确地反应出是否发送成功（而不需要在没发送成功时自动再次发送）
                   #####                 ，暂时关闭了此离线再发功能，也许以后的其它逻辑中用的上，但还是暂时取消了相关代码！！
                   #####                 【JS认为】桥接转发如不成功就应该立即由调用者知道，不然有此离线功能的话就会导致为何消息
                   #####                 报已发出，但就是没有收到的问题，这对调试和用户体验都不利！*/
                //// 尝试将之前因发送中遇到错误时暂存的消息再次进行发送
                //while (true) {
                //    // 取出数组第一个单元（并删除原数据中的该单元）
                //    var m = offlinePubQueue.shift();
                //    if (!m) {
                //        logger.debug('[3-IMMQ-↑] - [startPublisher()中] [___]在channel成功创建后，当前取出的数据第一个单元已为空，publish没有继续！[当前数组队列长度：%d]！【OK】',offlinePubQueue.length);
                //        break;
                //    }
                //
                //    logger.debug('[3-IMMQ-↑] - [startPublisher()中] [...]在channel成功创建后，正在publish消息 m[0]=%s、m[1]=%s,、m[2]=%s' +
                //        '，[当前数组队列长度：%d]！【OK】', m[0], m[1], m[2],offlinePubQueue.length);
                //    publish(m[0], m[1], m[2]);
                //}
            });
        });
    }
    else{
        logger.info("[3-IMMQ-↑] - [startPublisher()中] amqpConn是空的，本次startPublisher没有继续【NO】");
    }
}

/**
 * Method to publish a message.
 *
 * @param exchange
 * @param routingKey 队列名
 * @param content 消息内容
 * @param resultCallback
 */
function publish(exchange, routingKey, content, resultCallback) {
    try {
        pubChannel.publish(exchange, routingKey, content, { persistent: true },
            function(err, ok) {// 此回调中的ok在发送成功的情况下值是undefine，无视它吧

                if (err) {

                    /* ##### 20160919 by js：为了让publish调用时能准确地反应出是否发送成功（而不需要在没发送成功时自动再次发送）
                     #####                 ，暂时关闭了此离线再发功能，也许以后的其它逻辑中用的上，但还是暂时取消了相关代码！！
                     #####                 【JS认为】桥接转发如不成功就应该立即由调用者知道，不然有此离线功能的话就会导致为何消息
                     #####                 报已发出，但就是没有收到的问题，这对调试和用户体验都不利！*/
                    //// 放回数组尾部位，以便下次正常时接之前的失败时间顺序被Publish出去，从而保证消息顺序
                    ////offlinePubQueue.unshift([exchange, routingKey, content]);// 推到数组首部
                    //offlinePubQueue.push([exchange, routingKey, content]);// 推到数组尾部
                    //logger.error('[3-IMMQ-↑] - [startPublisher()中] publish()时出错了，原因：%s【数据[%s,%s,%s]' +
                    //    '已重新放回数组首位，数组新长度：%d】, pubChannel马上会被关闭。', JSON.stringify(err), exchange
                    //    , routingKey, content, offlinePubQueue.length);

                    logger.error('[3-IMMQ-↑] - [startPublisher()中] publish()时出错了，原因：%s【数据[%s,%s,%s]' +
                        '将由调用者进行处理（离线存起来？）】, pubChannel马上会被关闭。', JSON.stringify(err), exchange
                        , routingKey, content);
                    pubChannel.connection.close();

                    // publish 结果通过回调通知调用者
                    resultCallback(false);
                }
                else{
                    logger.info('[3-IMMQ-↑] - [startPublisher()中] publish()成功了(ok=%s) ！(数据:%s,%s,%s)【OK】'
                        , ok, exchange, routingKey, content);

                    // publish 结果通过回调通知调用者
                    resultCallback(true);
                }
            });
    } catch (e) {

        /* ##### 20160919 by js：为了让publish调用时能准确地反应出是否发送成功（而不需要在没发送成功时自动再次发送）
         #####                 ，暂时关闭了此离线再发功能，也许以后的其它逻辑中用的上，但还是暂时取消了相关代码！！
         #####                 【JS认为】桥接转发如不成功就应该立即由调用者知道，不然有此离线功能的话就会导致为何消息
         #####                 报已发出，但就是没有收到的问题，这对调试和用户体验都不利！*/
        // 放回数组尾部位，以便下次正常时接之前的失败时间顺序被Publish出去，从而保证消息顺序
        ////offlinePubQueue.unshift([exchange, routingKey, content]);// 推到数组首部
        //offlinePubQueue.push([exchange, routingKey, content]);// 推到数组尾部
        //logger.error('[3-IMMQ-↑] - [startPublisher()中] publish()时Exception了，原因：%s【数据[%s,%s,%s]已重新放回数组首位' +
        //    '，当前数组长度：%d】', JSON.stringify(e), exchange, routingKey, content, offlinePubQueue.length);

        logger.error('[3-IMMQ-↑] - [startPublisher()中] publish()时Exception了，原因：%s【数据[%s,%s,%s]将由调用者进' +
            '行处理（离线存起来？）】', JSON.stringify(e), exchange, routingKey, content);

        // publish 结果通过回调通知调用者
        resultCallback(false);
    }
}

// A worker that acks messages only if processed succesfully
function startWorker() {
    amqpConn.createChannel(function(err, ch) {
        if (closeOnErr(err, ' ↓[startWorker()中的amqpConn.createChannel]')) return;

        ch.on("error", function(err) {
            //console.error("[AMQP] channel error", err.message);
            logger.error('[3-IMMQ-↓] - [startWorker()中] ch.on("error")事件触发了，err：%s ', JSON.stringify(err));

            // add by js: 20160922
            // 通过关闭连接从而触发Connection的close事件，而close事件将会重启整个MQProvider服务，从而达到重启startWorker()目的
            if (closeOnErr(err, ' - ↓[startWorker()中] ch.on("error")事件触发了]')) return;
        });

        ch.on("close", function() {
            //console.log("[AMQP] channel closed");
            logger.error('[3-IMMQ-↓] - [startWorker()中] ch.on("close")事件触发了，channel 关闭了！');
        });

        ch.prefetch(10);
        ch.assertQueue(config.im_mq_queue_app2web, { durable: true }, function(err, _ok) {
            if (closeOnErr(err, ' - ↓[[startWorker(config.im_mq_queue_app2web)中] 的ch.assertQueue]')) return;

            ch.consume(config.im_mq_queue_app2web, processMsg, { noAck: false });
            //console.log("Worker is started");
            logger.info('[3-IMMQ-↓] - [startWorker(config.im_mq_queue_app2web)中] Worker已经成功开启并运行中。【OK】');
        });

        function processMsg(msg) {
            work(msg, function(ok) {
                try {
                    if (ok)
                        ch.ack(msg);
                    else
                        ch.reject(msg, true);
                } catch (e) {
                    closeOnErr(e, ' - ↓[[startWorker(config.im_mq_queue_app2web)中] 的ch.consume(processMsg回调函数中)]');
                }
            });
        }
    });
}

function work(bufForProtocal, fn) {
    var strForProtocal = bufForProtocal.content.toString();
    logger.info("[3-IMMQ-↓] - [startWorker()中] Got msg：%s", strForProtocal);

    //cb(true);
    if(callBack_forWork){
        // fn 的参数就是回调函数callBack_forWork的返回值，将作为processMsg函数的work回调中匿名function(ok)的ok参数值
        fn(callBack_forWork(JSON.parse(strForProtocal)));
    }
    else{
        fn(false);
        logger.error("[3-IMMQ-↓] - [startWorker()中] 回调方函数 callBack_forWork 未设置，请应用层检查！【NO】");
    }
}

function closeOnErr(err, desc) {
    if (!err) return false;
    //console.error("[AMQP] error", err);
    logger.error('[3-IMMQ] 操作MQ连接时出现错误，DESC：%s, 错误信息：%s', desc, JSON.stringify(err));
    amqpConn.close();
    return true;
}



/**
 * 外部方法：用于设置MQ中间件收到桥接消息时的回调函数。
 *
 * @param fn fn的参数：是“fn(Protocal)”
 *           fn返回值：true表示此消息正确处理完成，false表示未成功处理（MQ会将原消息重放回队列以便下次再取到）
 *                    【注意】尽量不要返回false，不然消息被放回队列后容易死循环，出错作好记录备查即可！！
 */
exports.setCallBack_forWork = function(fn){
    callBack_forWork = fn;
};


/**
 * 外部方法：MQ中间件服务启动方法。
 *
 * @type {start}
 */
exports.start = start;


/**
 * 外部方法：发送一条跨机器消息。
 *
 * @param p 要发送的内容，Protocal对象
 * @param resultCallback 发送结果回调函数，形如：fn(result)，参数result为布尔值：true表示publish成功、否则表示publish失败
 */
exports.publishBridge = function(p, resultCallback){
    publish("", config.im_mq_queue_web2app, new Buffer(JSON.stringify(p)), resultCallback);
};

//exports.testPublish = function(){
//    setInterval(function() {
//        publish("", config.im_mq_queue_web2app, new Buffer('work work work 好 ！【'
//            +common.formatDate(new Date(), 'yyyy-MM-dd hh:mm:ss.S')+'】-from Nodejs')); // TODO 真正生产时，这个队列名要定一个正式的！
//    }, 10000);
//};

